package com.lecosa.flink;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.util.regex.Matcher;
import java.util.regex.Pattern;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

public class MySQLSink extends RichSinkFunction<String> {
    private Connection connection;
    private PreparedStatement preparedStatement;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        // 加载驱动程序
        Class.forName("com.mysql.cj.jdbc.Driver");
        // 获取数据库连接
        String url = "jdbc:mysql://192.168.249.112:3306/shufen";
        String user = "root";
        String password = "root";
        connection = DriverManager.getConnection(url, user, password);
        // 创建 PreparedStatement 对象
        String sql = "replace INTO student1 (id,no, name, gender) VALUES (?,?, ?, ?)";
        System.out.println(sql);
        preparedStatement = connection.prepareStatement(sql);
    }

    @Override
    public void invoke(String value, Context context) throws Exception {

        String id = "";
        String name = "";
        String no = "";
        String gender = "";

        // 匹配 id 的正则表达式
        Pattern idPattern = Pattern.compile("id=(\\d+)");
        Matcher idMatcher = idPattern.matcher(value);
        if (idMatcher.find()) {
            id = idMatcher.group(1);
        }


        // 匹配 no 的正则表达式
        Pattern noPattern = Pattern.compile("no=([^,]+)");
        Matcher noMatcher = noPattern.matcher(value);
        if (noMatcher.find()) {
            no = noMatcher.group(1);
        }

        // 匹配 name 的正则表达式
        Pattern namePattern = Pattern.compile("name=([^,]+)");
        Matcher nameMatcher = namePattern.matcher(value);
        if (nameMatcher.find()) {
            name = nameMatcher.group(1);
        }

        Pattern genderPattern = Pattern.compile("gender=([^,]+)");
        Matcher genderMatcher = namePattern.matcher(value);
        if (genderMatcher.find()) {
            gender = nameMatcher.group(1);
        }


        // 将值插入到新表中
        preparedStatement.setString(1, id);
        preparedStatement.setString(2, no);
        preparedStatement.setString(3, name);
        preparedStatement.setString(4, gender);
        preparedStatement.executeUpdate();
    }

    @Override
    public void close() throws Exception {
        super.close();
        // 关闭连接和 PreparedStatement
        if (preparedStatement != null) {
            preparedStatement.close();
        }
        if (connection != null) {
            connection.close();
        }
    }
}
